// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: taekwon

#pragma once

#include <algorithm>
#include <atomic>
#include <memory>
#include <string>
#include <vector>

#include <fmt/format.h>
#include <boost/algorithm/string/erase.hpp>
#include <boost/algorithm/string/replace.hpp>

#include "coin/proto/coin_order_gateway.pb.h"
#include "coin/proto/coin_request.pb.h"
#include "coin2/base/hostname_util.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/executor/topic_publisher.h"
#include "orio/io/base.h"
#include "orio/io/concat.h"
#include "orio/record/topic.h"

struct ProtoLogReturn {
  int64_t timestamp;
  std::string log_content;
};

class BaseProtoLogger {
 public:
  BaseProtoLogger(const std::string& log_root, const std::string& machine)
      : log_root_(ProcessLogRoot(log_root)),
        machine_(ProcessMachine(machine)),
        expiry_timestamp_(0),
        log_timestamp_(0),
        seq_(0),
        muted_((log_root_ == "/dev/null") ? true : false) {
    buf_.reserve(32768);
  }

  virtual ~BaseProtoLogger() = default;

  void EnsureWriter(int64_t timestamp) {
    if (writer_ && timestamp >= expiry_timestamp_) {
      InitializeWriter(timestamp);
    } else if (!writer_) {
      InitializeWriter(timestamp);
    }
  }

  void InitializeWriter(int64_t timestamp) {
    constexpr int64_t refresh_period_ns = 1800'000'000'000ll;
    expiry_timestamp_ = (timestamp / refresh_period_ns + 1) * refresh_period_ns;
    log_timestamp_ = timestamp;
    auto ensured_path = GetEnsuredLogPath(timestamp);
    writer_.reset(orio::record::TopicWriter::ToFile(ensured_path));
    if (init_callback_) {
      init_callback_(timestamp);
    }
  }

  void SetInitCallbackFunc(std::function<void(int64_t)> callback) { init_callback_ = callback; }

  virtual std::string GetEnsuredLogPath(int64_t timestamp) const = 0;

  void Write(const std::string& pbuf_str, bool flush, int64_t timestamp) {
    if (muted_) return;
    EnsureWriter(timestamp);
    buf_.resize(orio::record::TopicRecord::struct_size_with_data(pbuf_str.size()));
    orio::record::TopicRecord* r = reinterpret_cast<orio::record::TopicRecord*>(buf_.data());
    uint64_t seq = GetNextSeq();

    r->data_len = pbuf_str.size();
    r->queue_seq = seq;
    r->topic_id = 0;
    r->topic_seq = seq;
    r->timestamp = timestamp;
    const std::byte* data = reinterpret_cast<const std::byte*>(pbuf_str.data());
    std::copy(data, data + r->data_len, r->data);

    CHECK(writer_) << "no way it can't be initialized";
    writer_->Write(*r);
    if (flush) {
      Flush();
    }
  }

  void Flush() {
    CHECK(writer_) << "no way it can't be initialized";
    writer_->Flush();
  }

  // TODO(taekwon): move to another place
  static bool EnsureDirExists(const std::string& dir) {
    if (std::experimental::filesystem::exists(dir)) {
      return true;
    }
    return std::experimental::filesystem::create_directories(dir);
  }

  static std::string GetDateTime(int64_t timestamp) {
    auto time_str = Iso8601FromTimestamp(timestamp).substr(0, 19);
    boost::erase_all(time_str, "-");
    boost::erase_all(time_str, ":");
    boost::replace_all(time_str, "T", "-");
    return time_str;
  }

  int64_t expiry_timestamp() const { return expiry_timestamp_; }
  int64_t log_timestamp() const { return log_timestamp_; }

 private:
  uint64_t GetNextSeq() { return seq_++; }

 protected:
  const std::string log_root_;
  const std::string machine_;

 private:
  static std::string ProcessLogRoot(const std::string& log_root) { return ExpandUser(log_root); }

  static std::string ProcessMachine(const std::string& machine) {
    if (machine.empty()) {
      return GetLocalHostname();
    } else {
      return machine;
    }
  }

 private:
  std::unique_ptr<orio::record::TopicWriter> writer_;
  int64_t expiry_timestamp_;
  int64_t log_timestamp_;
  std::string buf_;
  std::function<void(int64_t)> init_callback_;
  std::atomic<uint64_t> seq_;
  bool muted_;
};

// originally from coin2/flow/reader.h
class BaseProtoReader {
 public:
  BaseProtoReader(
      const std::string& root_dir,
      const std::string& machine,
      int64_t begin_timestamp,
      int64_t end_timestamp,
      size_t max_record_size)
      : buf_(new char[max_record_size]), max_record_size_(max_record_size), curr_read_idx_(0) {}
  virtual ~BaseProtoReader() = default;

  bool ReadRaw(ProtoLogReturn* out) {
    if (!reader_) {
      if (curr_read_idx_ >= num_files_) {
        return false;
      }
      try {
        reader_ = std::unique_ptr<orio::record::TopicReader>(
            orio::record::TopicReader::FromFile(files_[curr_read_idx_], max_record_size_));
      } catch (std::exception& e) {
        LOG(WARNING) << "skip " << files_[curr_read_idx_] << " " << e.what();
        curr_read_idx_++;
        return ReadRaw(out);
      }
      curr_read_idx_++;
    }
    orio::record::TopicRecord* record = reinterpret_cast<orio::record::TopicRecord*>(buf_.get());
    try {
      if (reader_->Read(record)) {
        out->log_content = std::string(
            reinterpret_cast<char*>(record->data),
            reinterpret_cast<char*>(record->data) + record->data_len);
        out->timestamp = record->timestamp;
        return true;
      } else {
        reader_ = nullptr;
        return ReadRaw(out);
      }
    } catch (std::exception& e) {
      reader_ = nullptr;
      return ReadRaw(out);
    }
    return false;
  }

  virtual void SetFileList(
      const std::string& root_dir,
      const std::string& machine,
      int64_t begin_timestamp,
      int64_t end_timestamp /* exclusive */) = 0;

 protected:
  std::vector<std::string> files_;
  size_t num_files_;

 private:
  std::unique_ptr<char[]> buf_;
  std::unique_ptr<orio::record::TopicReader> reader_;
  const int max_record_size_;
  size_t curr_read_idx_;
};
